# !/usr/bin/env python
# -*- coding: utf-8 -*-
# @File  : 异步确认机制（简化版）.py
# @Author: dongguangwen
# @Date  : 2025-04-12 15:42
import pika
import uuid

unconfirmed_set = set()


def on_open(connection):
    print("[✔] 连接已打开")
    connection.channel(on_channel_open)


def on_channel_open(channel):
    print("[✔] 通道已打开")
    global ch
    ch = channel

    # 设置消息确认回调
    ch.confirm_delivery(on_confirm_callback)

    # 声明交换机、队列、绑定
    ch.exchange_declare(exchange='async_confirm_ex', exchange_type='direct', durable=True)
    ch.queue_declare(queue='async_confirm_q', durable=True)
    ch.queue_bind(queue='async_confirm_q', exchange='async_confirm_ex', routing_key='async_key')

    # 消息发送放在队列绑定后执行
    connection.ioloop.call_later(1, send_messages)


def on_confirm_callback(method_frame):
    if method_frame.method.NAME == 'Basic.Ack':
        print(f"[✅] 消息确认 delivery_tag={method_frame.method.delivery_tag}")
    elif method_frame.method.NAME == 'Basic.Nack':
        print(f"[❌] 消息未确认 delivery_tag={method_frame.method.delivery_tag}")


def send_messages():
    for i in range(5):
        message = f"异步消息 {i}"
        properties = pika.BasicProperties(
            message_id=str(uuid.uuid4()),
            delivery_mode=2  # 消息持久化
        )
        ch.basic_publish(
            exchange='async_confirm_ex',
            routing_key='async_key',
            body=message,
            properties=properties
        )
        print(f"[→] 已发送：{message}")

    # 可选：5 秒后关闭连接
    connection.ioloop.call_later(5, stop_connection)


def stop_connection():
    print("[ℹ️] 关闭连接")
    connection.close()


# ========== 启动 ==========

parameters = pika.ConnectionParameters(
    '192.168.1.119',
    port=5672,
    virtual_host='my_vhost',
    credentials=pika.PlainCredentials('admin', 'password'),
    socket_timeout=10,  # 增加超时时间
    connection_attempts=3,
    retry_delay=3
)

# 🔥 使用 SelectConnection 必须使用 ioloop
connection = pika.SelectConnection(parameters, on_open_callback=on_open)

try:
    print("[🔁] 开始事件循环")
    connection.ioloop.start()
except KeyboardInterrupt:
    connection.close()
    connection.ioloop.start()
